package com.atguigu.userprofile.ml.pipeline

import org.apache.spark.ml.classification.{Classifier, DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, StringIndexerModel, VectorAssembler, VectorIndexer}
import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.DataFrame

class MyPipeline {


  // 流水线对象
  private var pipeline:Pipeline=null
  // 流水线训练后模型
  private var pipelineModel:PipelineModel=null


  //// 以下为参数   在MyPipeline创建时可以传递进来////////////////////
  //最大分类树（用于识别连续值特征和分类特征）
  private var maxCategories=5
  // 最大分支数
  private var maxBins=5
  // 最大树深度
  private var maxDepth=5
  //最小分支包含数据条数
  private var minInstancesPerNode=1
  //最小分支信息增益
  private var minInfoGain=0.0

  def setMaxDepth(maxDepth: Int): MyPipeline ={
      this.maxDepth=maxDepth
      this
  }

 // 标签列和特征列
  var labelColumnName:String=null
  var featureColumnNames:Array[String]=null

// 初始化流水线
  def  init(): MyPipeline ={
       pipeline = new Pipeline().setStages(
         Array( createLabelInderer(),
                createFeatureAssemble(),
                createFeatureIndexer(),
                createClassifier()) )
     this
  }

  def setLabelColumnName (labelColumnName:String): MyPipeline ={
    this.labelColumnName=labelColumnName
    this
  }

  def setFeatureColumnNames (featureColumnNames:Array[String]): MyPipeline ={
    this.featureColumnNames=featureColumnNames
    this
  }

  // 训练
  def train(dataFrame:DataFrame): Unit ={
      pipelineModel  = pipeline.fit(dataFrame)
  }

  //预测 //会增加一列数据 prediction
  def  predict(dataFrame:DataFrame): DataFrame = {
    val predictedDataFrame: DataFrame = pipelineModel.transform(dataFrame)
    predictedDataFrame
  }

  //  创建流水线
  //   四大组件
  //    1  标签索引
  //    2  特征聚合
   //   3 特征索引
   //  4  分类器
  //   训练方法
  //    预测方法
  //   评估方法
  //   模型储存方法
  //   模型加载方法
  //    把预测结果向量还原为原值
  //    打印决策树  (训练后）
  //    查看特征的权重 (训练后）

  //    1  标签索引
  //      作用 ：   把“参考答案”，做成矢量值，相当于把字符串变为矢量值
  //        男-->0  女-->1  xxx->2 ...   按照 原值出现的频率从大到小 --》 矢量从小到大分配置  出现频率越高，矢量值越小
  //    input 放数据标签列的名称
  //   output 为新生产的标签矢量列命名
  def  createLabelInderer(): StringIndexer ={
         val stringIndexer = new StringIndexer()
           .setInputCol(labelColumnName).setOutputCol("label_index")
//           .setHandleInvalid("skip")
         stringIndexer
  }

  //    2  特征聚合
  //  作用 ：  标识哪些列是特征列
  //    input : 特征列名的数组
  // output 为新生产的特征集合列命名
  def createFeatureAssemble(): VectorAssembler ={
         val vectorAssembler = new VectorAssembler()
           .setInputCols(featureColumnNames)
           .setOutputCol("feature_assemble")
           .setHandleInvalid("skip")
    vectorAssembler
  }


  //   3 特征索引
  //   作用： 把离散型特征变为矢量值
  //    连续型特征不变
  //   input : 固定 feature_assemble
  //   output :  命名为 feature_index
  //   maxCategories ： 最大分类数  默认是5 超过该值 视为连续值特征，低于等于视为离散值特征
  // HandleInvalid : skip跳过非法值 ，默认抛异常
  def  createFeatureIndexer(): VectorIndexer ={
       val vectorIndexer = new VectorIndexer()
    vectorIndexer.setInputCol("feature_assemble")
      .setOutputCol("feature_index")
      .setMaxCategories(maxCategories)
      .setHandleInvalid("skip")
    vectorIndexer
  }


  //  4  分类器
  //  labelCol ： 标签矢量列
  // FeaturesCol:  特征矢量列
   // PredictionCol: 命名预测结果列（矢量值) 固定 prediction
   //  impurity : 特征度量算法 entropy 或者 gini    默认gini
  // MaxDepth :   树最大深度  默认5
  //  Maxbins :  最大某一个特征产生的分支数   默认32
  // minInstancesPerNode :  某一个分支最少的数据条数  默认1
  // MinInfoGain : 最小信息获取   gini 系数减值 或者 熵减值  默认 0
  def   createClassifier(): DecisionTreeClassifier ={
         val decisionTreeClassifier = new  DecisionTreeClassifier()
           .setLabelCol("label_index")
           .setFeaturesCol("feature_index")
           .setPredictionCol("prediction")
           .setImpurity("gini")
           .setMaxDepth(maxDepth)
           .setMaxBins(maxBins)
           .setMinInstancesPerNode(minInstancesPerNode)
           .setMinInfoGain(minInfoGain)
    decisionTreeClassifier
  }

  //    打印决策树  (训练后）
  def  getDecisionTreeString(): String ={
        if( pipelineModel ==null) throw  new RuntimeException("needing traning first !!!! ")
        val decisionTreeClassificationModel: DecisionTreeClassificationModel = pipelineModel.stages(3).asInstanceOf[DecisionTreeClassificationModel]
        decisionTreeClassificationModel.toDebugString
  }


  //    查看特征的权重 (训练后）
  def  getFeatureImportances(): String ={
    if( pipelineModel ==null) throw  new RuntimeException("needing traning first !!!! ")
    val decisionTreeClassificationModel: DecisionTreeClassificationModel = pipelineModel.stages(3).asInstanceOf[DecisionTreeClassificationModel]
     decisionTreeClassificationModel.featureImportances.toString
  }

  //   评估方法
  def  evaluate(dataFrame: DataFrame): Unit ={
    val predictionLabelRdd: RDD[(Double, Double)] = dataFrame.rdd.map { row =>
      (row.getAs[Double]("prediction"), row.getAs[Double]("label_index"))
    }

     val metrics = new MulticlassMetrics(predictionLabelRdd )
      println("准确率："+metrics.accuracy)

      metrics.labels.foreach{label=>
        println(s" $label :召回率->${metrics.recall(label)}")
        println(s" $label :精确率->${metrics.precision(label)}")
      }

    dataFrame.createTempView("predict_table")
    dataFrame.sparkSession.sql(
      """ select  tp,fn, fp,tn,tt,ct,
        |         tp/(tp+fp) pRate_male,
        |         tp/(tp+fn) rRate_male,
        |         tt/ct   aRate  from (
        |   select sum(if(label_index=1.0 and  prediction=1.0 ,1,0)) tp ,
        |       sum(if(label_index=1.0 and  prediction=0.0 ,1,0))  fn ,
        |       sum(if(label_index=0.0 and  prediction=1.0 ,1,0))    fp ,
        |       sum(if(label_index=0.0 and  prediction=0.0 ,1,0)) tn ,
        |       sum(if(label_index=prediction ,1,0)) tt ,
        |       count(*) ct
        |       from predict_table
        |   ) pt
      """.stripMargin ).show(100,false)



  }

  // InputCol 数据的预测矢量结果列
  // Output  转换为原值的新列

   def convertOriginLabel(dataFrame: DataFrame): DataFrame ={

     if( pipelineModel ==null) throw  new RuntimeException("needing traning first !!!! ")
        val stringIndexerModel: StringIndexerModel = pipelineModel.stages(0).asInstanceOf[StringIndexerModel]
      val indexToString = new IndexToString()
      indexToString.setLabels(stringIndexerModel.labels)
     indexToString.setInputCol("prediction")
     indexToString.setOutputCol("prediction_origin")
     val convertedDataFrame: DataFrame = indexToString.transform(dataFrame)
     convertedDataFrame
   }

  //存储
  def save(path:String): Unit ={
      this.pipelineModel.write.overwrite().save(path)
  }

//  加载
  def load(path:String): MyPipeline ={
      this.pipelineModel = PipelineModel.load(path)
      this
  }


}
